package com.li.flink.datastream

import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

object WindowApp {
  def main(args: Array[String]): Unit = {
    //获取运行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //连接socket
    val text = env.socketTextStream("127.0.0.1", 9001)
    //处理数据
    //    val wordCount: DataStream[Integer] = text.map(m => {
    //      Integer.parseInt(m)
    //    })
    //    wordCount.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5))).sum(0).print()

    //    val wordCount: DataStream[Tuple2[String, Int]] = text.map(m => {
    //      val arr = m.split(",")
    //      (arr(0), arr(1).toInt)
    //    })
    //    wordCount.keyBy(_._1).window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
    //      .sum(1).print()

    //    val reduceFunction = new ReduceFunction[Tuple2[String, Int]] {
    //      override def reduce(t: (String, Int), t1: (String, Int)): (String, Int) = {
    //        (t._1, t1._2 + t._2)
    //      }
    //    }
    //    val wordCount: DataStream[Tuple2[String, Int]] = text.map(m => {
    //      val arr = m.split(",")
    //      (arr(0), arr(1).toInt)
    //    })
    //    wordCount.keyBy(_._1)
    //      .window(TumblingProcessingTimeWindows.of(Time.seconds(20)))
    //      .reduce(reduceFunction).print()


    class MyProcessWindowFunction extends ProcessWindowFunction[(String, Int), String, String, TimeWindow] {
      def process(key: String, context: Context, input: Iterable[(String, Int)], out: Collector[String]) = {
        var max = 0;
        input.foreach(i => {
          if (i._2 > max) {
            max = i._2
          }
        })
        out.collect("max value" + max)
      }
    }

    val wordCount: DataStream[Tuple2[String, Int]] = text.map(m => {
      val arr = m.split(",")
      (arr(0), arr(1).toInt)
    })
    wordCount.keyBy(_._1)
      .window(TumblingProcessingTimeWindows.of(Time.seconds(20)))
      .process(new MyProcessWindowFunction()).print()

    
    env.execute("WindowApp")
  }
}
